package org.jflame.context.mq;

import org.jflame.commons.util.StringHelper;
import org.jflame.context.cache.CacheWrapper;

import java.io.IOException;
import java.sql.SQLException;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;

import com.rabbitmq.client.Channel;

/**
 * 实现重复消息判断的消息接收监听器父类.
 * <p>
 * 通过缓存消息id,判断消息是否重复. 默认是将消息id放入缓存中实现
 * 
 * @author charles.zhang
 */
public abstract class AbstractRabbitMessageListener {

    protected final Logger logger = LoggerFactory.getLogger(getClass());
    /**
     * 缓存前缀
     */
    protected String cacheKeyPrefix;

    protected CacheWrapper<String,Integer> cacheWrapper;

    public AbstractRabbitMessageListener() {
    }

    public AbstractRabbitMessageListener(CacheWrapper<String,Integer> cache) {
        this.cacheWrapper = cache;
        cacheKeyPrefix = getClass().getName();
    }

    public enum MqAckAction {
        ACCEPT, // 处理成功
        RETRY, // 可以重试消息
        REJECT, // 无需重试消息,丢弃
    }

    public void onMessage(Message message, Channel channel) throws Exception {
        String msgId = message.getMessageProperties()
                .getMessageId();
        String cacheKey = null;
        if (StringHelper.isNotEmpty(msgId)) {
            cacheKey = cacheKeyPrefix + msgId;
            if (isDuplicateMsg(cacheKey)) {
                logger.warn("重复的消息:{}", msgId);
                replyMq(cacheKey, message, channel, MqAckAction.REJECT);
                return;
            }
        }
        MqAckAction reply = null;
        try {
            reply = handleMessage(message.getBody(), message.getMessageProperties());
        } catch (Exception e) {
            logger.error("消息处理异常,消息ID:{},ex:{}", msgId, e);
            reply = exceptionHandle(e);
        }
        replyMq(cacheKey, message, channel, reply);
    }

    private void replyMq(String cacheKey, Message message, Channel channel, MqAckAction reply) throws IOException {
        long deliveryTag = message.getMessageProperties()
                .getDeliveryTag();
        switch (reply) {
            case ACCEPT:
                channel.basicAck(deliveryTag, false);
                storeMsgId(cacheKey);
                break;
            case RETRY:
                channel.basicNack(deliveryTag, false, true);// 重新入队
                storeMsgId(cacheKey);
                break;
            case REJECT:
                channel.basicReject(deliveryTag, false);// 丢弃
                break;
            default:
                channel.basicAck(deliveryTag, false);
                storeMsgId(cacheKey);
                break;
        }
    }

    /**
     * 判断消息是否重复
     * 
     * @param msgId
     * @return
     */
    protected boolean isDuplicateMsg(String msgId) {
        return cacheWrapper.get(msgId) != null;
    }

    /**
     * 消息id存入
     * 
     * @param msgId
     */
    protected void storeMsgId(String msgId) {
        cacheWrapper.put(msgId, 1);
    }

    /**
     * 根据异常类型确定应答本条消息的行为
     * 
     * @param ex
     * @return
     */
    protected MqAckAction exceptionHandle(Throwable ex) {
        if (ex.getCause() instanceof SQLException) {
            return MqAckAction.REJECT;
        }
        return MqAckAction.ACCEPT;
    }

    /**
     * 消息处理业务逻辑
     * 
     * @param message
     * @param messageProperties
     * @return MqAckAction
     * @throws Exception
     */
    protected abstract MqAckAction handleMessage(final byte[] message, MessageProperties messageProperties)
            throws Exception;
}
